package org.codehaus.activemq.service.impl;

import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import javax.jms.JMSException;
import org.codehaus.activemq.broker.BrokerClient;
import org.codehaus.activemq.broker.impl.BrokerClientImpl;
import org.codehaus.activemq.message.ActiveMQMessage;
import org.codehaus.activemq.message.DefaultWireFormat;
import org.codehaus.activemq.message.MessageAck;
import org.codehaus.activemq.message.Packet;
import org.codehaus.activemq.message.WireFormat;
import org.codehaus.activemq.service.TransactionTask;
import org.codehaus.activemq.util.JMSExceptionHelper;

public abstract class PacketTransactionTask
  implements TransactionTask, Externalizable
{
  private static final transient WireFormat wireFormat = new DefaultWireFormat();
  private BrokerClient brokerClient;
  private Packet packet;

  public static TransactionTask fromBytes(byte[] data)
    throws IOException
  {
    Packet packet = wireFormat.fromBytes(data);
    return createTask(packet);
  }

  public byte[] toBytes() throws JMSException, IOException {
    return wireFormat.toBytes(this.packet);
  }

  public static TransactionTask readTask(ObjectInput in) throws IOException {
    Packet packet = readPacket(in);
    return createTask(packet);
  }

  public static TransactionTask createTask(Packet packet) throws IOException {
    if ((packet instanceof MessageAck)) {
      return new MessageAckTransactionTask(null, (MessageAck)packet);
    }
    if ((packet instanceof ActiveMQMessage)) {
      return new SendMessageTransactionTask(null, (ActiveMQMessage)packet);
    }

    throw new IOException("Unexpected packet type: " + packet);
  }

  public static void writeTask(TransactionTask task, ObjectOutput out) throws IOException
  {
    if ((task instanceof PacketTransactionTask)) {
      PacketTransactionTask packetTask = (PacketTransactionTask)task;
      writePacket(packetTask.getPacket(), out);
    }
    else {
      out.writeObject(task);
    }
  }

  protected PacketTransactionTask(BrokerClient brokerClient, Packet packet) {
    this.brokerClient = brokerClient;
    this.packet = packet;
  }

  public Packet getPacket() {
    return this.packet;
  }

  public void writeExternal(ObjectOutput out) throws IOException {
    writePacket(this.packet, out);
  }

  public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
    this.packet = readPacket(in);
  }

  protected BrokerClient createBrokerClient(String consumerId)
    throws JMSException
  {
    BrokerClientImpl answer = new BrokerClientImpl();
    return answer;
  }

  protected BrokerClient getBrokerClient(String consumerId) throws JMSException
  {
    this.brokerClient = createBrokerClient(consumerId);

    return this.brokerClient;
  }

  protected static void writePacket(Packet packet, ObjectOutput out) throws IOException {
    try {
      wireFormat.writePacket(packet, out);
    }
    catch (JMSException e) {
      throw JMSExceptionHelper.newIOException(e);
    }
  }

  protected static Packet readPacket(ObjectInput in) throws IOException {
    return wireFormat.readPacket(in);
  }
}